package ad

import java.util
import java.util.Date

import kafka.common.TopicAndPartition
import kafka.message.MessageAndMetadata
import kafka.serializer.StringDecoder
import kafka.utils.{ZKGroupTopicDirs, ZkUtils}
import org.I0Itec.zkclient.ZkClient
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.Row
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka.{HasOffsetRanges, KafkaUtils, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import sessionanalyze.conf.ConfigurationManager
import sessionanalyze.constant.Constants
import sessionanalyze.dao.factory.DAOFactory
import sessionanalyze.domain.{AdBlacklist, AdClickTrend, AdProvinceTop3, AdStat}
import sessionanalyze.util.DateUtils

import scala.collection.mutable.{ArrayBuffer, ListBuffer}





/*
                    .::::.
                  .::::::::.
                 :::::::::::
             ..:::::::::::'	  FUCK YOU
           '::::::::::::'		Goddess bless, never BUG
             .::::::::::
        '::::::::::::::..
             ..::::::::::::.
           ``::::::::::::::::
            ::::``:::::::::'        .:::.
           ::::'   ':::::'       .::::::::.
         .::::'      ::::     .:::::::'::::.
        .:::'       :::::  .:::::::::' ':::::.
       .::'        :::::.:::::::::'      ':::::.
      .::'         ::::::::::::::'         ``::::.
  ...:::           ::::::::::::'              ``::.
 ```` ':.          ':::::::::'                  ::::..
                    '.:::::'                    ':'````..
                    
 ━━━━━━━━━━━━━━━━━━━━ 女神保佑,永无BUG ━━━━━━━━━━━━━━━━━━━━
*/
object AdMockReal {
  def main(args: Array[String]): Unit = {
    //指定组名
    val group = "g001"
    //创建SparkConf
    val conf = new SparkConf().setAppName(Constants.SPARK_APP_NAME_AD).setMaster("local[2]")
    //创建SparkStreaming，并设置间隔时间
    val ssc = new StreamingContext(conf, Seconds(5))
    //指定消费的 topic 名字
    val topic = ConfigurationManager.getProperty(Constants.KAFKA_TOPICS)
    //指定kafka的broker地址(sparkStream的Task直连到kafka的分区上，用更加底层的API消费，效率更高)
    val brokerList = ConfigurationManager.getProperty(Constants.KAFKA_METADATA_BROKER_LIST)

    //指定zk的地址，后期更新消费的偏移量时使用(以后可以使用Redis、MySQL来记录偏移量)
    val zkQuorum = "hadoop2:2181,hadoop3:2181,hadoop4:2181"
    //创建 stream 时使用的 topic 名字集合，SparkStreaming可同时消费多个topic
    val topics: Set[String] = Set(topic)

    //创建一个 ZKGroupTopicDirs 对象,其实是指定往zk中写入数据的目录，用于保存偏移量
    val topicDirs = new ZKGroupTopicDirs(group, topic)
    //获取 zookeeper 中的路径 "/g001/offsets/wordcount/"
    val zkTopicPath = s"${topicDirs.consumerOffsetDir}"

    //准备kafka的参数
    val kafkaParams = Map(
      "metadata.broker.list" -> brokerList,
      "group.id" -> group,
      //从头开始读取数据
      "auto.offset.reset" -> kafka.api.OffsetRequest.SmallestTimeString
    )

    //zookeeper 的host 和 ip，创建一个 client,用于跟新偏移量量的
    //是zookeeper的客户端，可以从zk中读取偏移量数据，并更新偏移量
    val zkClient = new ZkClient(zkQuorum)

    //查询该路径下是否字节点（默认有字节点为我们自己保存不同 partition 时生成的）
    //zkTopicPath  -> /g001/offsets/wordcount/
    val children = zkClient.countChildren(zkTopicPath)

    var kafkaStream: InputDStream[(String, String)] = null

    //如果 zookeeper 中有保存 offset，我们会利用这个 offset 作为 kafkaStream 的起始位置
    var fromOffsets: Map[TopicAndPartition, Long] = Map()

    //如果保存过 offset
    if (children > 0) {
      for (i <- 0 until children) {
        // /g001/offsets/wordcount/0/10001

        // /g001/offsets/wordcount/0
        val partitionOffset = zkClient.readData[String](s"$zkTopicPath/${i}")
        // wordcount/0
        val tp = TopicAndPartition(topic, i)
        //将不同 partition 对应的 offset 增加到 fromOffsets 中
        // wordcount/0 -> 10001
        fromOffsets += (tp -> partitionOffset.toLong)
      }
      //Key: kafka的key   values: "hello tom hello jerry"
      //这个会将 kafka 的消息进行 transform，最终 kafak 的数据都会变成 (kafka的key, message) 这样的 tuple
      val messageHandler = (mmd: MessageAndMetadata[String, String]) => (mmd.key(), mmd.message())

      //通过KafkaUtils创建直连的DStream（fromOffsets参数的作用是:按照前面计算好了的偏移量继续消费数据）
      //[String, String, StringDecoder, StringDecoder,     (String, String)]
      //  key    value    key的解码方式   value的解码方式
      kafkaStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder, (String, String)](ssc, kafkaParams, fromOffsets, messageHandler)
    } else {
      //如果未保存，根据 kafkaParam 的配置使用最新(largest)或者最旧的（smallest） offset
      kafkaStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)
    }

    //偏移量的范围
    var offsetRanges = Array[OffsetRange]()

    //直连方式只有在KafkaDStream的RDD中才能获取偏移量，那么就不能到调用DStream的Transformation
    //所以只能子在kafkaStream调用foreachRDD，获取RDD的偏移量，然后就是对RDD进行操作了
    //依次迭代KafkaDStream中的KafkaRDD
    kafkaStream.foreachRDD { kafkaRDD =>
      //只有KafkaRDD可以强转成HasOffsetRanges，并获取到偏移量
      offsetRanges = kafkaRDD.asInstanceOf[HasOffsetRanges].offsetRanges
      val lines: RDD[String] = kafkaRDD.map(_._2)

      //设置检查点
      ssc.checkpoint("checkpoint")

      //对RDD进行操作，触发Action
      val data: RDD[(String, String, String, String, String)] = lines.map(line => {
        //对字段进行切分
        val fields = line.split(" ")
        //日期
        val date = DateUtils.formatTime(new Date(fields(0).toLong))
        //省份
        val province = fields(1)
        //城市
        val city = fields(2)
        //用户Id
        val userId = fields(3)
        //广告Id
        val adId = fields(4)
        (date, province, city, userId, adId)
      })
      data.cache()
      //过滤出黑名单
      val filterList: List[String] = data.map(tup => ((tup._1.split(" ")(0),tup._5,tup._4),1L)).reduceByKey(_+_).filter(_._2>100).map(tup => (tup._1._3)).collect.toList
      //将其插入数据库
      insertAdBlacklist(filterList)



      //将黑名单集合转换为RDD
      val filters: RDD[(String, Boolean)] = ssc.sparkContext.parallelize(filterList).map((_,false))
      //进行黑名单过滤
      val filtered = data.map(tup => (tup._4,(tup._1.split(" ")(0),tup._2,tup._3,tup._5))).leftOuterJoin(filters).filter(_._2._2.getOrElse(true))
      //对过滤后的数据进行缓存
      filtered.cache()

      //对每天各省各市各广告点击统计
      val ad_stat: RDD[((String, String, String, String), Int)] = filtered.map(tup => (tup._2._1,1)).reduceByKey(_+_)
      insertAdStar(ad_stat)

      //统计每天各省各广告点击的Top3
      val adTop3: Array[((String, String, String), Int)] = filtered.map(tup => ((tup._2._1._1,tup._2._1._2,tup._2._1._4),1)).reduceByKey(_+_).sortBy(_._2,false).take(3)
      insertTop3(adTop3)

      val adClickTrend: RDD[((String, String, String, String), Long)] = data.map(tup => {
        val times = tup._1.split(" ")
        val date = times(0)
        val hour = times(1).split(":")(0)
        val minute = times(1).split(":")(1)
        val adId = tup._5
        ((date, hour, minute, adId), 1L)
      }).reduceByKey(_ + _)
      insertThread(adClickTrend)


//      lines.foreachPartition(partition =>
//        partition.foreach(x => {
//          println(x)
//        })
//      )

      for (o <- offsetRanges) {
        //  /g001/offsets/wordcount/0
        val zkPath = s"${topicDirs.consumerOffsetDir}/${o.partition}"
        //将该 partition 的 offset 保存到 zookeeper
        //  /g001/offsets/wordcount/0/20000
        ZkUtils.updatePersistentPath(zkClient, zkPath, o.untilOffset.toString)
      }
    }

    ssc.start()
    ssc.awaitTermination()




  }

  def insertThread(value: RDD[((String, String, String, String), Long)])={
    val DAO = DAOFactory.getAdClickTrendDAO
    value.foreachPartition(it =>{
      val list = new util.ArrayList[AdClickTrend]()
      it.foreach(tup => {
        val ad = new AdClickTrend
        ad.setDate(tup._1._1)
        ad.setHour(tup._1._2)
        ad.setMinute(tup._1._3)
        ad.setAdid(tup._1._4.toLong)
        ad.setClickCount(tup._2)
        list.add(ad)
      })
      DAO.updateBatch(list)
    })
    println("更新数据ad_click_trend表完成")

  }

  def insertTop3(tuples: Array[((String, String, String), Int)])={
    val DAO = DAOFactory.getAdProvinceTop3DAO
    val list = new util.ArrayList[AdProvinceTop3]()
    for(tup <- tuples){
      val ad = new AdProvinceTop3
      ad.setDate(tup._1._1)
      ad.setProvince(tup._1._2)
      ad.setAdid(tup._1._3.toLong)
      ad.setClickCount(tup._2)
      list.add(ad)
    }
    DAO.updateBatch(list)
    println("更新数据表ad_province_top3成功")
  }

  def insertAdStar(value: RDD[((String, String, String, String), Int)])={
    val adStatDAO = DAOFactory.getAdStatDAO
    value.foreachPartition(it => {
      val list = new util.ArrayList[AdStat]()
      it.map(tup => {
        val ad = new AdStat
        ad.setDate(tup._1._1)
        ad.setProvince(tup._1._2)
        ad.setCity(tup._1._3)
        ad.setAdid(tup._1._4.toLong)
        ad.setClickCount(tup._2)
        list.add(ad)
      })
      adStatDAO.updateBatch(list)
    })
    println("更新数据表Ad_Stat成功")
  }

  //将黑名单插入数据库
  def insertAdBlacklist(value: List[String])={
    val adDao = DAOFactory.getAdBlacklistDAO
    val list = new util.ArrayList[AdBlacklist]()
    value.foreach(x => {val ad = new AdBlacklist();ad.setUserid(x.toLong);list.add(ad)})
    adDao.insertBatch(list)
    println("更新数据表Ad_Blacklist成功")
  }
}
